package com.walkud.rxjava.transforming;

import com.walkud.rxjava.utils.PrintLogSubscriber;
import com.walkud.rxjava.utils.RxGroupBySubscriber;
import com.walkud.rxjava.utils.RxSubscribe;
import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.internal.operators.observable.ObservableGroupBy;
import io.reactivex.observables.GroupedObservable;

import java.util.List;
import java.util.Map;

import static com.walkud.rxjava.utils.DataUtil.getArrayData;
import static com.walkud.rxjava.utils.DataUtil.getGroupData;

/**
 * 变换操作符简单示例
 * Buffer — periodically gather items from an Observable into bundles and emit these bundles rather than emitting the items one at a time
 * FlatMap — transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable
 * GroupBy — divide an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key
 * Map — transform the items emitted by an Observable by applying a function to each item
 * Scan — apply a function to each item emitted by an Observable, sequentially, and emit each successive value
 * Window — periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time
 */
public class TransformingSimple {

    public static void main(String[] args) {
//        buffer();
//        flatMap();
//        groupBy();
//        map();
//        scan();
        window();
    }

    /**
     * 定期将发射的数据变换为Observable被观察者，感觉类似于buffer操作符，在buffer之上加上了from操作符
     */
    private static void window() {
        Observable.fromArray(getArrayData()).window(3).subscribe(new RxSubscribe<Observable<String>>() {
            @Override
            public void call(Observable<String> stringObservable) {
                stringObservable.subscribe(PrintLogSubscriber.create());
            }
        });
    }

    /**
     * 将前一个发射的数据与当前数据进行变换处理并发射
     */
    private static void scan() {
        Observable.fromArray(getArrayData()).scan(new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                //这里与前一个字符串相连接
                return s + s2;
            }
        }).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 将发射数据类型变换为另一种类型
     */
    private static void map() {
        Observable.just(6).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer + " * 2 = " + (integer * 2);
            }
        }).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 将发射的数据变换为一个分组
     */
    private static void groupBy() {
        Observable observable = Observable.just(1, 2, 3, 3, 4, 2, 3, 5, 4, 3).groupBy(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer % 2 == 0 ? "偶数" : "奇数";
            }
        });

        //RxJava原生处理观察者(订阅者)
//        observable.subscribe(new Consumer<GroupedObservable<String, Integer>>() {
//            @Override
//            public void accept(GroupedObservable<String, Integer> groupedObservable) {
//                groupedObservable.subscribe(new Consumer<Integer>() {
//
//                    @Override
//                    public void accept(Integer integer) {
//                        System.out.println("Key:" + groupedObservable.getKey() + "," + integer);
//                    }
//                });
//            }
//        });


        //自定义分组处理观察者(订阅者)
        observable.subscribe(new RxGroupBySubscriber() {
            @Override
            public void call(Map<Object, List<Object>> map) {
                for (Map.Entry<Object, List<Object>> entry : map.entrySet()) {
                    System.out.println("Key:" + entry.getKey());
                    for (Object o : entry.getValue()) {
                        System.out.println("Value:" + o);
                    }
                }
            }
        });
    }

    /**
     * 将一个被观察者变换成另一个被观察者(常用)
     * 主要场景：用于一个请求依赖于另外一个请求(或多个请求)时使用
     */
    private static void flatMap() {
        Observable.just("Login").flatMap(new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(String s) throws Exception {
                return Observable.just(s + "，getUserInfo");
            }
        }).subscribe(PrintLogSubscriber.create());
    }


    /**
     * 将发射的数据变换成集合后进行发射
     */
    private static void buffer() {
        //将发射的数据打包成集合发射
        Observable.fromArray(getArrayData()).buffer(3).subscribe(new RxSubscribe<List<String>>() {
            @Override
            public void call(List<String> strings) {
            }
        });
    }


}
